import org.apache.spark.streaming.flume._
import org.apache.spark.streaming.StreamingContext
import java.net.InetSocketAddress
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming._
import org.apache.spark.streaming.flume._

object FlumePullWordCount
{


    def main(args: Array[String]): Unit =
    {


      if(args.length != 2)
      {
        System.err.println("Usage: FlumePullWordCount <hostname> <port>")
        System.exit(1)
      }


      val Array(hostname, port) = args


      val sparkConf = new SparkConf().setMaster("local[4]").setAppName("FlumePullWordCount")
      val ssc = new StreamingContext(sparkConf, Seconds(5))


      //TODO... 如何使用SparkStreaming整合Flume
      val flumeStream = FlumeUtils.createPollingStream(ssc, hostname, port.toInt)


      flumeStream.map(x=> new String(x.event.getBody.array()).trim)
        .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()


      ssc.start()
      ssc.awaitTermination()
    }
  }

